
//    网络反应器
#include "stdafx.h"
#include "NetworkReactor.h"


//    回调函数

//    关闭连接回调
void CloseEventCallbackFun(evutil_socket_t fd, short event, void *arg)
{
    ((CNetworkReactor*)arg)->CloseEventCallback(fd, event, arg);
}

//    监听回调
void ListenerEventCallbackFun(struct evconnlistener *listener, evutil_socket_t fd,
    struct sockaddr *sa, int socklen, void *user_data)
{
    ((CNetworkReactor*)user_data)->ListenerEventCallback(listener, fd, sa, socklen, user_data);
}

//    读回调(接收数据)
void ReadCallbackFun(struct bufferevent *bev, void *ctx)
{
    ((CNetworkIO*)ctx)->ReadCallback(bev, ctx);
}

//    写回调(发送数据)
void WriteCallbackFun(struct bufferevent *bev, void *ctx)
{
    ((CNetworkIO*)ctx)->WriteCallback(bev, ctx);
}

//    事件回调
void EventCallbackFun(struct bufferevent *bev, short what, void *ctx)
{
    ((CNetworkIO*)ctx)->EventCallback(bev, what, ctx);
}

//    错误事件回调
void ErrCallbackFun(int err)
{
#ifdef _DEBUG
    printf("错误:%d\n", err);
#endif
}


//    网络输入输出处理类

CNetworkIO::CNetworkIO()
    : _bev(NULL)
    , _reactor(NULL)
    , _type(0)
    , _listener_port(0)
    , _ref_count(1)//    默认为1，表示已经创建对象，但没引用
    , _off(false)
{
}

CNetworkIO::CNetworkIO(CNetworkIO &right)
    : _bev(NULL)
{
    _network_info = right._network_info;
    _reactor = right._reactor;
    _type = right._type;
    _listener_port = right._listener_port;
    _off = right._off;

    Process::CAutoLock    lock(&right._cs_ref_count);
    _ref_count = right._ref_count;
}

CNetworkIO::~CNetworkIO()
{
    if (_bev != NULL)
    {
        bufferevent_free(_bev);
        _bev = NULL;
        _off = true;
    }
}

//    向对应的网络写数据(自动产生crc校验)
bool CNetworkIO::write(NETWORK_DATA &network_data)
{
    if (_bev == NULL)
    {
        SetError("网络未连接!");
        return false;
    }

    if (network_data._len > PACKAGE_MAX_LEN)
    {    
        SetError("数据包太大!");
        return false;
    }

    //    生成crc校验
    network_data._crc32 = CSingleton<CCrc32>::Instance()->CalcCrc32FromBuf(network_data._buffer,sizeof(unsigned char) * network_data._len);

    //AutoBuffereventLock auto_buff_lock(_bev);
    evbuffer *output = NULL;
    output = bufferevent_get_output(_bev);

    //    发送包头
    if (evbuffer_add(output, &network_data, sizeof(NETWORK_DATA)) == -1)
    {
        SetError("[write]发送包头失败!");
        return false;
    }

    //    发送数据
    if (evbuffer_add(output, network_data._buffer, network_data._len) == -1)
    {
        SetError("[write]发送数据失败!");
        return false;
    }

    return true;
}

//    建立连接
bool CNetworkIO::Open(NETWORK_INFO network_info, bufferevent *bev, CNetworkReactor *reactor)
{
    _network_info = network_info;
    _bev = bev;
    _reactor = reactor;
    _type = 1;
    _listener_port = 0;

    bufferevent_setwatermark(_bev, EV_WRITE, 0, WRITE_BUFF_MAX);

    return true;
}

bool CNetworkIO::Open(NETWORK_INFO network_info, bufferevent *bev, CNetworkReactor *reactor, unsigned short listener_port)
{
    _network_info = network_info;
    _bev = bev;
    _reactor = reactor;
    _type = 0;
    _listener_port = listener_port;

    bufferevent_setwatermark(_bev, EV_WRITE, 0, WRITE_BUFF_MAX);

    return true;
}

// 关闭连接
bool CNetworkIO::Colse()
{
    SubRef();

    //    发送删除信号给反应堆
    _reactor->CloseNetworkIO();    
    return true;
}

CNetworkIO& CNetworkIO::operator=(CNetworkIO &right)
{
    _network_info = right._network_info;
    _reactor = right._reactor;
    _type = right._type;
    _listener_port = right._listener_port;
    _off = right._off;

    Process::CAutoLock    lock(&right._cs_ref_count);
    _ref_count = right._ref_count;
    return *this;
}

void CNetworkIO::WriteCallback(bufferevent *bev, void *user_data)
{
    //    写反馈,暂时不使用
}

void CNetworkIO::ReadCallback(bufferevent *bev, void *user_data)
{
    //    0.加锁
    //    1.寻找包头
    //    2.等待数据接收完毕
    //    3.验证数据
    //    4.如果出错从下一个字节重新寻找包头,在寻找到新包头后丢弃前面数据
    //    5.如果正常获取数据调用回调

    std::vector<NETWORK_DATA*> vec_network_data;
    
    //    自动锁作用域
    {
        //    0.加锁
        //AutoBuffereventLock auto_buff_lock(_bev);
        CCrc32 crc32;
        
        evbuffer_iovec *v = NULL;
        unsigned long flag = START_FLAG;

        evbuffer *input = bufferevent_get_input(_bev);

        if (evbuffer_get_length(input) <= 0)
        {
            return;
        }

        evbuffer_ptr start_ptr;
        evbuffer_ptr_set(input, &start_ptr, 0, EVBUFFER_PTR_SET);

        do
        {
            //    1.寻找包头标识
            start_ptr = evbuffer_search(input, (char*)&flag, sizeof(unsigned long), &start_ptr);
            if (start_ptr.pos == -1)
            {
                //    清除非规定包数据(在清除时预留4个字节,防止标识(flag)被清除)
                size_t buffer_len = evbuffer_get_length(input);
                if (buffer_len > 4)
                {
                    evbuffer_drain(input, buffer_len - 4);
                }
                break;
            }
            else
            {
                evbuffer_drain(input, start_ptr.pos);
                evbuffer_ptr_set(input, &start_ptr, 0, EVBUFFER_PTR_SET);
            }

            //    2.等待数据接收完毕
            //    获取包头
            if (evbuffer_get_length(input) < sizeof(NETWORK_DATA))
            {
                break;
            }

            NETWORK_DATA *p_head_network_data = (NETWORK_DATA*)evbuffer_pullup(input, sizeof(NETWORK_DATA));
            if (p_head_network_data == NULL)
            {
                break;
            }
            NETWORK_DATA head_network_data;
            memcpy(&head_network_data, p_head_network_data, sizeof(NETWORK_DATA));

            //    如果大于最大包长度，说明数据有误,继续寻找下一个包
            if (head_network_data._len > PACKAGE_MAX_LEN || head_network_data._len == 0)
            {
                if (evbuffer_ptr_set(input, &start_ptr, 1, EVBUFFER_PTR_ADD))
                    break;
                continue;
            }

            //    等待剩下的数据
            if (evbuffer_get_length(input) < sizeof(NETWORK_DATA) + head_network_data._len)
            {
                break;
            }

            //    3.验证数据
            crc32.Start();
            evbuffer_ptr ptr;
            //    移动指针ptr到数据部分
            if (evbuffer_ptr_set(input, &ptr, sizeof(NETWORK_DATA), EVBUFFER_PTR_SET))
            {
                break;
            }
            int n_vec = evbuffer_peek(input, head_network_data._len, &ptr, NULL, 0);
            v = new evbuffer_iovec[n_vec];
            n_vec = evbuffer_peek(input, head_network_data._len, &ptr, v, n_vec);
            unsigned long read_len = 0;//    接收到的数据大小
            for (int i = 0; i<n_vec; ++i)
            {
                size_t len = v[i].iov_len;
                if (read_len + len > head_network_data._len)
                {
                    len = head_network_data._len - read_len;
                }
                crc32.Set((unsigned char*)v[i].iov_base, len);
                read_len += len;
            }
            delete []v;
            v = NULL;

            if (crc32.Stop() != head_network_data._crc32)
            {
                if (evbuffer_ptr_set(input, &start_ptr, 1, EVBUFFER_PTR_ADD))
                    break;
                continue;
            }

            //    4.提取数据
            //    清除包中数据部分以前的数据
            if (evbuffer_drain(input, ptr.pos) == -1)
            {
                continue;
            }

            //    获取包数据部分
            head_network_data._buffer = new unsigned char[head_network_data._len];
            if (evbuffer_remove(input, head_network_data._buffer, head_network_data._len) != (int)head_network_data._len)
            {
                delete []head_network_data._buffer;
                head_network_data._buffer = NULL;
                continue;
            }
            
            NETWORK_DATA *p_network_data = new NETWORK_DATA();
            *p_network_data = head_network_data;
            vec_network_data.push_back(p_network_data);

            //    将搜索起始位置设置为evbuffer头部
            if (evbuffer_ptr_set(input, &start_ptr, 0, EVBUFFER_PTR_SET))
            {
                break;
            }

        }while(true);
    }
    
    
    //    5.如果正常获取数据调用回调
    for (unsigned int i = 0; i<vec_network_data.size(); ++i)
    {
        //    写接口回调
        read(vec_network_data[i]);

        //    回调完后释放对应内存
        if (vec_network_data[i] != NULL)
        {
            if (vec_network_data[i]->_buffer != NULL)
            {
                delete []vec_network_data[i]->_buffer;
                vec_network_data[i]->_buffer = NULL;
            }

            delete vec_network_data[i];
            vec_network_data[i] = NULL;
        }
    }
    vec_network_data.clear();
}

void CNetworkIO::EventCallback(bufferevent *bev, short events, void *user_data)
{
    if (bev != _bev)
    {
        return;
    }

    //    调用回调
    event(events);

    if (events & BEV_EVENT_EOF) //    连接断开
    {
        if (!_off)
        {
            Colse();
            _off = true;
        }
    } 
    else if (events & BEV_EVENT_ERROR) //    连接出错
    {
        if (!_off)
        {
            Colse();
            _off = true;
        }
    }
    else if (events & BEV_EVENT_TIMEOUT) //    连接超时
    {
        if (!_off)
        {
            Colse();
            _off = true;
        }
    }
}



//    网络反应器类

CNetworkReactor::CNetworkReactor()
    : _start(false)
    , _base(NULL)
{
    #ifdef WIN32
        WSADATA wsa_data;
        WSAStartup(0x0201, &wsa_data);
    #endif

    //    多线程安全
    evthread_use_windows_threads();    

    //    设置错误回调
    event_set_fatal_callback(ErrCallbackFun);

    //    创建基础事件
    _base = event_base_new();
    if (!_base) 
    {
        SetError("构造异常! event_base_new发生错误.");
        throw _err.c_str();
    }
}

CNetworkReactor::~CNetworkReactor()
{
    Stop();    
}

//    启动反应器
bool CNetworkReactor::Start()
{
    if (_start)
    {
        return true;
    }
    _start = true;

    //    创建基础事件
    if (_base == NULL)
    {
        _base = event_base_new();
        if (!_base) 
        {
            SetError("event_base_new发生错误.");
            _start = false;
            return false;
        }
    }    

    //    启动事件循环线程
    BeginThread();

    return true;
}

//    停止反应器
//    注意：所有的事件的删除必须在一次性事件中触发，
//    如果用其他线程调用在远端不停发送数据时会导致死锁
bool CNetworkReactor::Stop()
{
    if (_base == NULL)
    {
        return false;
    }

    if (!_start)
    {
        return true;
    }
    _start = false;

    //    释放监听事件
    {
        Process::CAutoLock    lock(&_cs_map_listener);
        std::map<evconnlistener*, LISTENER_INFO_BASE*>::iterator it_listener = _map_listener.begin();

        for(;it_listener != _map_listener.end(); ++it_listener)
        {
            evconnlistener_free((*it_listener).first);
            if ((*it_listener).second != NULL)
            {
                delete (*it_listener).second;
                (*it_listener).second = NULL;
            }            
        }
        _map_listener.clear();
    }

    //    释放连接事件
    {
        Process::CAutoLock    lock(&_cs_map_network_io);
        std::map<NETWORK_INFO, CNetworkIO*>::iterator it_network = _map_network_io.begin();

        for(;it_network != _map_network_io.end(); ++it_network)
        {
            if ((*it_network).second != NULL)
            {
                (*it_network).second->SubRef();
            }            
        }
        CloseNetworkIO();
    }

    //    停止事件循环
    struct timeval delay = { 2, 0 };
    event_base_loopexit(_base, &delay);    

    //    停止循环
    EndThread();

    //    释放基础事件
    if (_base != NULL)
    {
        event_base_free(_base);
        _base = NULL;
    }

    return true;
}

//    删除连接事件
//    删除对应端口监听，及对应的连接事件
//    注意：所有的事件的删除必须在一次性事件中触发，
//    如果用其他线程调用在远端不停发送数据时会导致死锁
bool CNetworkReactor::DelConnectEvent(unsigned short port)
{
    //    删除监听事件
    {
        Process::CAutoLock    lock(&_cs_map_listener);
        std::map<evconnlistener*, LISTENER_INFO_BASE*>::iterator it_listener = _map_listener.begin();

        for(;it_listener != _map_listener.end(); ++it_listener)
        {
            if ((*it_listener).second != NULL && (*it_listener).second->_port == port)
            {
                evconnlistener_free((*it_listener).first);
                delete (*it_listener).second;
                (*it_listener).second = NULL;
                _map_listener.erase(it_listener);
                break;
            }
        }
    }

    //    删除连接事件
    {
        Process::CAutoLock    lock(&_cs_map_network_io);
        std::map<NETWORK_INFO, CNetworkIO*>::iterator it_network = _map_network_io.begin();

        while (it_network != _map_network_io.end())
        {
            if ((*it_network).second != NULL && (*it_network).second->GetListerPort() == port)
            {
                (*it_network).second->SubRef();
            }
            ++it_network;
        }
        CloseNetworkIO();
    }

    return true;
}

//    删除连接事件
//    注意：所有的事件的删除必须在一次性事件中触发，
//    如果用其他线程调用在远端不停发送数据时会导致死锁
bool CNetworkReactor::DelConnectEvent(std::string ip, unsigned short port)
{
    NETWORK_INFO network_info;
    network_info._ip = ip;
    network_info._family = AF_INET;
    network_info._port = port;

    //    是否找到
    bool is_find = false;
    {
        Process::CAutoLock    lock(&_cs_map_network_io);
        std::map<NETWORK_INFO, CNetworkIO*>::iterator it_network = _map_network_io.find(network_info);

        if (it_network != _map_network_io.end())
        {
            (*it_network).second->SubRef();
            is_find = true;
        }
    }

    if (is_find)
    {
        CloseNetworkIO();
    }

    return true;
}

//    线程函数
unsigned int CNetworkReactor::ThreadEntry(void)
{
    while (!terminated_)
    {
        if (_map_network_io.size() == 0 && _map_listener.size() == 0)
        {
            Sleep(2000);
            continue;
        }

        //    启动事件循环(此函数将阻塞)
        event_base_dispatch(_base);
    }

    return 0;
}

CNetworkReactor::CNetworkReactor(const CNetworkReactor &right)
{
    *this = right;
}

CNetworkReactor& CNetworkReactor::operator=(const CNetworkReactor &right)
{
    assert(0);        //    不能赋值!
    return *this;
}

//    发送一个网络关闭事件
bool CNetworkReactor::CloseNetworkIO()
{
    struct timeval tv;
    evutil_timerclear(&tv);
    tv.tv_sec = 1;
    event_base_once(_base,-1, EV_TIMEOUT, CloseEventCallbackFun, this, &tv);
    return true;
}

//    关闭网络连接事件回调(Network中_bev为空的都关闭)
bool CNetworkReactor::CloseEventCallback(evutil_socket_t fd, short event, void *arg)
{    
    Process::CAutoLock    lock(&_cs_map_network_io);
    std::map<NETWORK_INFO, CNetworkIO*>::iterator it_network = _map_network_io.begin();
    while (it_network != _map_network_io.end())
    {
        if ((*it_network).second->GetRef() <= 0)
        {
            printf("IP:%s:%ld 关闭中.......................\n", (*it_network).first._ip.c_str(), (*it_network).first._port);
            delete (*it_network).second;
            (*it_network).second = NULL;
            _map_network_io.erase(it_network);
            it_network = _map_network_io.begin();
        }
        else
        {
            ++it_network;
        }
    }
    return true;
}

//    监听事件回调
bool CNetworkReactor::ListenerEventCallback(struct evconnlistener *listener, evutil_socket_t fd,struct sockaddr *sa, int socklen, void *user_data)
{
    //    获取远端IP信息
    sockaddr_in sin;
    memcpy(&sin, sa, sizeof(sin));
    NETWORK_INFO network_info;
    network_info._ip = inet_ntoa(sin.sin_addr);
    network_info._family = sin.sin_family;
    network_info._port = sin.sin_port;

    CNetworkIO *p_network_io = NULL;

    {
        Process::CAutoLock    lock_listener(&_cs_map_listener);
        //    在监听列表中查询参数
        std::map<evconnlistener*, LISTENER_INFO_BASE*>::iterator it_listener = _map_listener.find(listener);
        if (it_listener == _map_listener.end())
        {
            SetError("没有对应的监听事件.");
            return false;
        }    

        //    创建bufferevent
        bufferevent *bev;
        bev = bufferevent_socket_new(_base, fd, BEV_OPT_CLOSE_ON_FREE|BEV_OPT_DEFER_CALLBACKS | BEV_OPT_THREADSAFE);
        if (!bev)
        {
            SetError("[ListenerEventCallback] Error constructing bufferevent!");
            return false;
        }

        //    创建数据处理对象
        p_network_io = (CNetworkIO*)(*it_listener).second->DefaultAllocationNetworkIO();
        p_network_io->Open(network_info, bev, this, (*it_listener).second->_port);

        //    设置事件回调
        bufferevent_setcb(bev, ReadCallbackFun, WriteCallbackFun, EventCallbackFun, p_network_io);
        bufferevent_enable(bev, EV_WRITE | EV_READ);
    }    

    {
        //    加入网络事件处理列表
        Process::CAutoLock    lock_network(&_cs_map_network_io);
        _map_network_io[network_info] = p_network_io;
    }

    return true;
}